package ex55.mm.flow.impound.quote.common.redis.push;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import ex55.mm.flow.impound.quote.AggregateDataFacade;
import ex55.mm.flow.impound.quote.aggregate.ExchangeProduct;
import ex55.mm.flow.impound.quote.exchange.api.MarketDataDepth;
import ex55.mm.flow.impound.quote.exchange.api.Product;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.CollectionUtils;

import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Set;

public class Receiver {
    private static final Logger LOGGER = LoggerFactory.getLogger(Receiver.class);

    private final AggregateDataFacade aggregateDataFacade;

    public Receiver(AggregateDataFacade aggregateDataFacade) {
        this.aggregateDataFacade = aggregateDataFacade;
    }

    private Set<String> prodcutIds = Sets.newHashSet();

    public void receiveMessage(String message) {
        LOGGER.info("Received <" + message + ">");
        JSONObject jsonObject = JSONObject.parseObject(message);

        JSONObject payload = jsonObject.getJSONObject("payload");

        if (payload != null && !CollectionUtils.isEmpty(payload.keySet())) {
            Map<String, ExchangeProduct> exchangeProductMap = aggregateDataFacade.getData_ExchangeProduct();
            payload.keySet().forEach(symbol -> {
                JSONObject symbolEntry = payload.getJSONObject(symbol);
                if (symbolEntry != null && !CollectionUtils.isEmpty(symbolEntry.keySet())) {
                    symbolEntry.keySet().forEach(exchangeId -> {
                        String productId = exchangeId + "." + symbol;
                        prodcutIds.add(productId);
                        if (exchangeProductMap.containsKey(productId)) {
                            ExchangeProduct exchangeProduct = exchangeProductMap.get(exchangeId + "." + symbol);
                            JSONObject depthEntry = symbolEntry.getJSONObject(exchangeId);

                            MarketDataDepth marketDataDepth = new MarketDataDepth();
                            marketDataDepth.setProduct(new Product()
                                    .setSymbol(symbol)
                                    .setBaseAsset(exchangeProduct.getBaseAsset())
                                    .setQuoteAsset(exchangeProduct.getQuoteAsset()));

                            JSONArray bids = depthEntry.getJSONArray("bids");
                            if (bids != null && bids.size() > 0) {
                                List<MarketDataDepth.Entry> bidEntries = Lists.newArrayList();
                                bids.stream().limit(exchangeProduct.getDepth()).forEach(object -> {
                                    JSONObject bid = (JSONObject) object;
                                    bidEntries.add(
                                            new MarketDataDepth.Entry()
                                                    .setPrice(bid.getBigDecimal("price").stripTrailingZeros())
                                                    .setQuantity(bid.getBigDecimal("qty").stripTrailingZeros()));
                                });
                                marketDataDepth.setBids(bidEntries);
                            }

                            JSONArray asks = depthEntry.getJSONArray("asks");
                            if (asks != null && asks.size() > 0) {
                                List<MarketDataDepth.Entry> askEntries = Lists.newArrayList();
                                asks.stream().limit(exchangeProduct.getDepth()).forEach(object -> {
                                    JSONObject ask = (JSONObject) object;
                                    askEntries.add(
                                            new MarketDataDepth.Entry()
                                                    .setPrice(ask.getBigDecimal("price").stripTrailingZeros())
                                                    .setQuantity(ask.getBigDecimal("qty").stripTrailingZeros()));
                                });
                                marketDataDepth.setAsks(askEntries);
                            }
                            marketDataDepth.setRefreshedAt(new Date(depthEntry.getLong("updateAt")));
                            marketDataDepth.setUpdatedAt(new Date(depthEntry.getLong("updateAt")));

                            aggregateDataFacade.insertMarketDataDepth(exchangeProduct.getProductId(), marketDataDepth);
                        }
                    });
                }
            });
        }
    }
}
